using System.Net;
using System.Net.Sockets;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using Microsoft.Extensions.Logging;
using ReactiveChannelMessaging.Core;

namespace ReactiveChannelMessaging.Communication;

/// <summary>
/// 响应式TCP通讯管理器
/// </summary>
public class ReactiveTcpCommunication : IAsyncDisposable
{
    private readonly ILogger<ReactiveTcpCommunication> _logger;
    private readonly ReactiveMessageBus _messageBus;
    private readonly Subject<TcpDataMessage> _receivedDataSubject;
    private readonly Subject<TcpStatusMessage> _statusSubject;
    private readonly List<IDisposable> _subscriptions;
    private readonly CancellationTokenSource _cancellationTokenSource;
    
    private TcpClient? _tcpClient;
    private TcpListener? _tcpListener;
    private NetworkStream? _networkStream;
    private readonly Dictionary<string, ClientConnection> _connectedClients;
    private Task? _listenTask;

    public string? RemoteEndPoint { get; private set; }
    public int Port { get; }
    public bool IsServer { get; }
    public bool IsConnected => _tcpClient?.Connected == true || _tcpListener != null;

    public ReactiveTcpCommunication(
        int port,
        bool isServer,
        ILogger<ReactiveTcpCommunication> logger,
        ReactiveMessageBus messageBus,
        string? remoteHost = null)
    {
        Port = port;
        IsServer = isServer;
        RemoteEndPoint = isServer ? null : remoteHost ?? "localhost";
        
        _logger = logger;
        _messageBus = messageBus;
        _receivedDataSubject = new Subject<TcpDataMessage>();
        _statusSubject = new Subject<TcpStatusMessage>();
        _subscriptions = new List<IDisposable>();
        _cancellationTokenSource = new CancellationTokenSource();
        _connectedClients = new Dictionary<string, ClientConnection>();

        SetupMessageSubscriptions();
    }

    /// <summary>
    /// 接收到的TCP数据流
    /// </summary>
    public IObservable<TcpDataMessage> ReceivedData => _receivedDataSubject.AsObservable();

    /// <summary>
    /// TCP连接状态变化流
    /// </summary>
    public IObservable<TcpStatusMessage> StatusChanges => _statusSubject.AsObservable();

    /// <summary>
    /// 已连接的客户端流（仅服务器模式）
    /// </summary>
    public IObservable<string> ConnectedClients => 
        _statusSubject.Where(s => s.Status == TcpConnectionStatus.ClientConnected)
                     .Select(s => s.ClientId ?? "");

    /// <summary>
    /// 启动TCP连接（客户端模式）或监听（服务器模式）
    /// </summary>
    public async Task<bool> StartAsync()
    {
        try
        {
            if (IsServer)
            {
                return await StartServerAsync();
            }
            else
            {
                return await StartClientAsync();
            }
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Failed to start TCP communication");
            await PublishStatusMessage(TcpConnectionStatus.Error, ex.Message);
            return false;
        }
    }

    /// <summary>
    /// 停止TCP连接
    /// </summary>
    public async Task StopAsync()
    {
        try
        {
            _cancellationTokenSource.Cancel();

            // 断开所有客户端连接
            foreach (var client in _connectedClients.Values.ToList())
            {
                await DisconnectClient(client.Id);
            }
            _connectedClients.Clear();

            // 关闭TCP客户端
            if (_tcpClient != null)
            {
                _networkStream?.Close();
                _tcpClient.Close();
                _tcpClient = null;
                _networkStream = null;
            }

            // 关闭TCP服务器
            if (_tcpListener != null)
            {
                _tcpListener.Stop();
                _tcpListener = null;
            }

            // 等待监听任务完成
            if (_listenTask != null)
            {
                await _listenTask;
                _listenTask = null;
            }

            await PublishStatusMessage(TcpConnectionStatus.Disconnected, "TCP communication stopped");
            _logger.LogInformation("TCP communication stopped");
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Error stopping TCP communication");
        }
    }

    /// <summary>
    /// 发送数据
    /// </summary>
    public async Task<bool> SendAsync(byte[] data, string? clientId = null)
    {
        try
        {
            if (IsServer)
            {
                return await SendToClientAsync(data, clientId);
            }
            else
            {
                return await SendToServerAsync(data);
            }
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Failed to send TCP data");
            return false;
        }
    }

    /// <summary>
    /// 发送字符串数据
    /// </summary>
    public async Task<bool> SendStringAsync(string data, string? clientId = null, System.Text.Encoding? encoding = null)
    {
        encoding ??= System.Text.Encoding.UTF8;
        var bytes = encoding.GetBytes(data);
        return await SendAsync(bytes, clientId);
    }

    /// <summary>
    /// 广播数据到所有客户端（仅服务器模式）
    /// </summary>
    public async Task<int> BroadcastAsync(byte[] data)
    {
        if (!IsServer)
        {
            _logger.LogWarning("Broadcast is only available in server mode");
            return 0;
        }

        var successCount = 0;
        var clients = _connectedClients.Values.ToList();
        
        foreach (var client in clients)
        {
            if (await SendToClientAsync(data, client.Id))
            {
                successCount++;
            }
        }

        _logger.LogInformation("Broadcasted data to {SuccessCount}/{TotalCount} clients", 
            successCount, clients.Count);
        
        return successCount;
    }

    private async Task<bool> StartServerAsync()
    {
        _tcpListener = new TcpListener(IPAddress.Any, Port);
        _tcpListener.Start();
        
        await PublishStatusMessage(TcpConnectionStatus.Listening, $"Server listening on port {Port}");
        _logger.LogInformation("TCP server started on port {Port}", Port);

        _listenTask = Task.Run(async () => await AcceptClientsAsync(_cancellationTokenSource.Token));
        
        return true;
    }

    private async Task<bool> StartClientAsync()
    {
        if (string.IsNullOrEmpty(RemoteEndPoint))
        {
            throw new InvalidOperationException("Remote endpoint is required for client mode");
        }

        _tcpClient = new TcpClient();
        await PublishStatusMessage(TcpConnectionStatus.Connecting, $"Connecting to {RemoteEndPoint}:{Port}");
        
        await _tcpClient.ConnectAsync(RemoteEndPoint, Port);
        _networkStream = _tcpClient.GetStream();
        
        await PublishStatusMessage(TcpConnectionStatus.Connected, $"Connected to {RemoteEndPoint}:{Port}");
        _logger.LogInformation("TCP client connected to {RemoteEndPoint}:{Port}", RemoteEndPoint, Port);

        // 启动数据接收任务
        _ = Task.Run(async () => await ReceiveDataAsync(_networkStream, "Server", _cancellationTokenSource.Token));
        
        return true;
    }

    private async Task AcceptClientsAsync(CancellationToken cancellationToken)
    {
        try
        {
            while (!cancellationToken.IsCancellationRequested && _tcpListener != null)
            {
                var tcpClient = await _tcpListener.AcceptTcpClientAsync();
                var clientEndPoint = tcpClient.Client.RemoteEndPoint?.ToString() ?? "Unknown";
                var clientId = Guid.NewGuid().ToString();
                
                var clientConnection = new ClientConnection(clientId, tcpClient, clientEndPoint);
                _connectedClients[clientId] = clientConnection;
                
                await PublishStatusMessage(TcpConnectionStatus.ClientConnected, 
                    $"Client {clientEndPoint} connected", clientId);
                
                _logger.LogInformation("Client {ClientEndPoint} connected with ID {ClientId}", 
                    clientEndPoint, clientId);

                // 为每个客户端启动数据接收任务
                _ = Task.Run(async () => 
                {
                    try
                    {
                        await ReceiveDataAsync(clientConnection.TcpClient.GetStream(), 
                            clientId, cancellationToken);
                    }
                    finally
                    {
                        await DisconnectClient(clientId);
                    }
                });
            }
        }
        catch (ObjectDisposedException)
        {
            // TCP监听器已关闭，正常退出
        }
        catch (Exception ex)
        {
            if (!cancellationToken.IsCancellationRequested)
            {
                _logger.LogError(ex, "Error accepting TCP clients");
            }
        }
    }

    private async Task ReceiveDataAsync(NetworkStream stream, string clientId, CancellationToken cancellationToken)
    {
        var buffer = new byte[4096];
        
        try
        {
            while (!cancellationToken.IsCancellationRequested && stream.CanRead)
            {
                var bytesRead = await stream.ReadAsync(buffer, 0, buffer.Length, cancellationToken);
                
                if (bytesRead == 0)
                {
                    // 连接已关闭
                    break;
                }

                var data = new byte[bytesRead];
                Array.Copy(buffer, data, bytesRead);

                var dataMessage = new TcpDataMessage(
                    Guid.NewGuid(),
                    DateTime.UtcNow,
                    "TcpCommunication",
                    clientId,
                    data,
                    TcpDataDirection.Received
                );

                _receivedDataSubject.OnNext(dataMessage);
                await _messageBus.PublishAsync(dataMessage);

                _logger.LogDebug("Received {BytesRead} bytes from client {ClientId}", 
                    bytesRead, clientId);
            }
        }
        catch (Exception ex)
        {
            if (!cancellationToken.IsCancellationRequested)
            {
                _logger.LogError(ex, "Error receiving data from client {ClientId}", clientId);
            }
        }
    }

    private async Task<bool> SendToServerAsync(byte[] data)
    {
        if (_networkStream == null || !_networkStream.CanWrite)
        {
            _logger.LogWarning("Cannot send data - not connected to server");
            return false;
        }

        try
        {
            await _networkStream.WriteAsync(data, 0, data.Length);
            await _networkStream.FlushAsync();

            var dataMessage = new TcpDataMessage(
                Guid.NewGuid(),
                DateTime.UtcNow,
                "TcpCommunication",
                "Server",
                data,
                TcpDataDirection.Sent
            );

            await _messageBus.PublishAsync(dataMessage);
            _logger.LogDebug("Sent {Length} bytes to server", data.Length);
            
            return true;
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Failed to send data to server");
            return false;
        }
    }

    private async Task<bool> SendToClientAsync(byte[] data, string? clientId)
    {
        if (string.IsNullOrEmpty(clientId))
        {
            // 发送到第一个连接的客户端
            clientId = _connectedClients.Keys.FirstOrDefault();
            if (string.IsNullOrEmpty(clientId))
            {
                _logger.LogWarning("No clients connected to send data to");
                return false;
            }
        }

        if (!_connectedClients.TryGetValue(clientId, out var client))
        {
            _logger.LogWarning("Client {ClientId} not found", clientId);
            return false;
        }

        try
        {
            var stream = client.TcpClient.GetStream();
            if (!stream.CanWrite)
            {
                _logger.LogWarning("Cannot write to client {ClientId}", clientId);
                return false;
            }

            await stream.WriteAsync(data, 0, data.Length);
            await stream.FlushAsync();

            var dataMessage = new TcpDataMessage(
                Guid.NewGuid(),
                DateTime.UtcNow,
                "TcpCommunication",
                clientId,
                data,
                TcpDataDirection.Sent
            );

            await _messageBus.PublishAsync(dataMessage);
            _logger.LogDebug("Sent {Length} bytes to client {ClientId}", data.Length, clientId);
            
            return true;
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Failed to send data to client {ClientId}", clientId);
            await DisconnectClient(clientId);
            return false;
        }
    }

    private async Task DisconnectClient(string clientId)
    {
        if (_connectedClients.TryGetValue(clientId, out var client))
        {
            try
            {
                client.TcpClient.Close();
                _connectedClients.Remove(clientId);
                
                await PublishStatusMessage(TcpConnectionStatus.ClientDisconnected, 
                    $"Client {client.EndPoint} disconnected", clientId);
                
                _logger.LogInformation("Client {ClientId} ({EndPoint}) disconnected", 
                    clientId, client.EndPoint);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error disconnecting client {ClientId}", clientId);
            }
        }
    }

    private void SetupMessageSubscriptions()
    {
        // 监听来自消息总线的TCP命令
        var tcpCommandSubscription = _messageBus.CommandMessages
            .Where(cmd => cmd.Command.StartsWith("tcp_"))
            .Subscribe(async cmd =>
            {
                try
                {
                    await ProcessTcpCommand(cmd);
                }
                catch (Exception ex)
                {
                    _logger.LogError(ex, "Error processing TCP command: {Command}", cmd.Command);
                }
            });

        _subscriptions.Add(tcpCommandSubscription);
    }

    private async Task ProcessTcpCommand(CommandMessage command)
    {
        var response = command.Command.ToLowerInvariant() switch
        {
            "tcp_send" => await HandleSendCommand(command),
            "tcp_send_string" => await HandleSendStringCommand(command),
            "tcp_broadcast" => await HandleBroadcastCommand(command),
            "tcp_status" => HandleStatusCommand(),
            "tcp_stop" => await HandleStopCommand(),
            "tcp_start" => await HandleStartCommand(),
            "tcp_list_clients" => HandleListClientsCommand(),
            _ => new { Success = false, Error = $"Unknown TCP command: {command.Command}" }
        };

        var responseMessage = new ResponseMessage(
            Guid.NewGuid(),
            DateTime.UtcNow,
            "TcpCommunication",
            command.Id,
            response.GetType().GetProperty("Success")?.GetValue(response) as bool? ?? false,
            response
        );

        await _messageBus.PublishAsync(responseMessage);
    }

    private async Task<object> HandleSendCommand(CommandMessage command)
    {
        if (command.Parameters?.TryGetValue("data", out var data) == true)
        {
            var clientId = command.Parameters.TryGetValue("clientId", out var id) ? id?.ToString() : null;
            
            if (data is byte[] bytes)
            {
                var success = await SendAsync(bytes, clientId);
                return new { Success = success, BytesSent = bytes.Length, ClientId = clientId };
            }
        }
        return new { Success = false, Error = "Invalid data parameter" };
    }

    private async Task<object> HandleSendStringCommand(CommandMessage command)
    {
        if (command.Parameters?.TryGetValue("text", out var text) == true)
        {
            var clientId = command.Parameters.TryGetValue("clientId", out var id) ? id?.ToString() : null;
            var success = await SendStringAsync(text.ToString() ?? "", clientId);
            return new { Success = success, TextSent = text, ClientId = clientId };
        }
        return new { Success = false, Error = "Invalid text parameter" };
    }

    private async Task<object> HandleBroadcastCommand(CommandMessage command)
    {
        if (!IsServer)
        {
            return new { Success = false, Error = "Broadcast only available in server mode" };
        }

        if (command.Parameters?.TryGetValue("data", out var data) == true)
        {
            byte[] bytes = data switch
            {
                byte[] b => b,
                string s => System.Text.Encoding.UTF8.GetBytes(s),
                _ => System.Text.Encoding.UTF8.GetBytes(data.ToString() ?? "")
            };

            var count = await BroadcastAsync(bytes);
            return new { Success = true, ClientsReached = count, BytesSent = bytes.Length };
        }
        return new { Success = false, Error = "Invalid data parameter" };
    }

    private object HandleStatusCommand()
    {
        return new
        {
            Success = true,
            Port = Port,
            IsServer = IsServer,
            IsConnected = IsConnected,
            RemoteEndPoint = RemoteEndPoint,
            ConnectedClients = _connectedClients.Count,
            ClientList = _connectedClients.Values.Select(c => new { c.Id, c.EndPoint }).ToList()
        };
    }

    private async Task<object> HandleStopCommand()
    {
        await StopAsync();
        return new { Success = true, Message = "TCP communication stopped" };
    }

    private async Task<object> HandleStartCommand()
    {
        var success = await StartAsync();
        return new { Success = success, Message = success ? "TCP communication started" : "Failed to start" };
    }

    private object HandleListClientsCommand()
    {
        if (!IsServer)
        {
            return new { Success = false, Error = "List clients only available in server mode" };
        }

        var clients = _connectedClients.Values.Select(c => new
        {
            c.Id,
            c.EndPoint,
            ConnectedAt = c.ConnectedAt,
            IsConnected = c.TcpClient.Connected
        }).ToList();

        return new { Success = true, Clients = clients, Count = clients.Count };
    }

    private async Task PublishStatusMessage(TcpConnectionStatus status, string description, string? clientId = null)
    {
        var statusMessage = new TcpStatusMessage(
            Guid.NewGuid(),
            DateTime.UtcNow,
            "TcpCommunication",
            Port,
            status,
            description,
            clientId
        );

        _statusSubject.OnNext(statusMessage);
        await _messageBus.PublishAsync(statusMessage);
    }

    public async ValueTask DisposeAsync()
    {
        await StopAsync();
        
        foreach (var subscription in _subscriptions)
        {
            subscription.Dispose();
        }
        
        _receivedDataSubject.Dispose();
        _statusSubject.Dispose();
        _cancellationTokenSource.Dispose();
    }
}

/// <summary>
/// 客户端连接信息
/// </summary>
public class ClientConnection
{
    public string Id { get; }
    public TcpClient TcpClient { get; }
    public string EndPoint { get; }
    public DateTime ConnectedAt { get; }

    public ClientConnection(string id, TcpClient tcpClient, string endPoint)
    {
        Id = id;
        TcpClient = tcpClient;
        EndPoint = endPoint;
        ConnectedAt = DateTime.UtcNow;
    }
}

/// <summary>
/// TCP数据消息
/// </summary>
public record TcpDataMessage(
    Guid Id,
    DateTime Timestamp,
    string Source,
    string ClientId,
    byte[] Data,
    TcpDataDirection Direction
) : ReactiveChannelMessaging.Core.Message(Id, Timestamp, Source);

/// <summary>
/// TCP状态消息
/// </summary>
public record TcpStatusMessage(
    Guid Id,
    DateTime Timestamp,
    string Source,
    int Port,
    TcpConnectionStatus Status,
    string Description,
    string? ClientId = null
) : ReactiveChannelMessaging.Core.Message(Id, Timestamp, Source);

/// <summary>
/// TCP数据方向
/// </summary>
public enum TcpDataDirection
{
    Received,
    Sent
}

/// <summary>
/// TCP连接状态
/// </summary>
public enum TcpConnectionStatus
{
    Disconnected,
    Connecting,
    Connected,
    Listening,
    ClientConnected,
    ClientDisconnected,
    Error
}